/**
 * Copyright © 2022 The TyIotCloud Authors: TangZhenzhi
 */
package org.mqtt.aes.server;

import com.alibaba.fastjson.JSONObject;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttVersion;
import io.netty.util.concurrent.Future;
import lombok.extern.slf4j.Slf4j;
import org.algorithm.Hmac;
import org.mqtt.*;

import java.nio.charset.StandardCharsets;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Slf4j
public class MqttIntegrationTest {

    static final String MQTT_HOST = "172.21.32.34";

    static final Integer MQTT_PORT = 32004;

//    static final String MQTT_HOST = "localhost";
//
//    static final Integer MQTT_PORT = 1884;
    static final int KEEPALIVE_TIMEOUT_SECONDS = 1000;
    static final ByteBufAllocator ALLOCATOR = new UnpooledByteBufAllocator(false);

    static EventLoopGroup eventLoopGroup = new NioEventLoopGroup(50);

    private static MqttClient initClient(String clientid, String username, String password) throws Exception {
        MqttClientConfig config = new MqttClientConfig();
        config.setTimeoutSeconds(KEEPALIVE_TIMEOUT_SECONDS);
        config.setClientId(clientid);
        config.setUsername(username);
        config.setPassword(password);
        config.setProtocolVersion(MqttVersion.MQTT_5);
        MqttClient client = MqttClient.create(config, null);
        client.setEventLoop(eventLoopGroup);
        Future<MqttConnectResult> connectFuture = client.connect(MQTT_HOST, MQTT_PORT);

        MqttConnectResult result;
        try {
            result = connectFuture.get(10, TimeUnit.SECONDS);
        } catch (TimeoutException ex) {
            ex.printStackTrace();
            connectFuture.cancel(true);
            client.disconnect();
            throw new RuntimeException(String.format("Failed to connect to MQTT server at %s.", 1111));
        }
        if (!result.isSuccess()) {
            connectFuture.cancel(true);
            client.disconnect();
            throw new RuntimeException(String.format("Failed to connect to MQTT server at %s. Result code is: %s", 1111, result.getReturnCode()));
        }
        return client;
    }

    public static void main(String[] args) throws Exception {
        List<ClientConfig> configs = getDeviceList();
        int i =0;
        for (ClientConfig config :configs) {
            try{
                String random = String.valueOf(System.currentTimeMillis()).substring(5,13);
                CipherConfig cipherConfig = new CipherConfig(config.getDeviceId(),"3",random);
                config.setCipherConfig(cipherConfig);
                String password = Hmac.hmacSha256(cipherConfig.getNonce(),config.getPassword());
                MqttCiphertextHandler handler = new MqttCiphertextHandler(cipherConfig.getClientId(),config.getDeviceId(),password);
                handler.setDeviceSecret(config.getPassword());
                config.setHandler(handler);
                MqttClient client = initClient(cipherConfig.getClientId(),config.getDeviceId(),password);
                client.on("/tyiotcloud/encrypt/random", new MqttHandler() {
                    @Override
                    public void onMessage(String topic, ByteBuf payload) {
                        byte[] data = new byte[payload.readableBytes()];
                        payload.readBytes(data);
                        try {
                            String dd = new
                                    String(config.getHandler().decrypt(data));
                            handler.getCipherConfig().setServerRandom(dd);
                            handler.reInit();
                            System.out.println("==============================="+dd);
                        }catch (Exception e){
                            e.printStackTrace();
                        }


                    }
                });
                client.on("ED718/downRaw/+/ts/data", new MqttHandler() {
                    @Override
                    public void onMessage(String topic, ByteBuf payload) {
                        byte[] data = new byte[payload.readableBytes()];
                        payload.readBytes(data);
                        try {
                            String dd = new
                                    String(config.getHandler().decrypt(data));
                            System.out.println("==============================="+dd);
                        }catch (Exception e){
                            e.printStackTrace();
                        }


                    }
                });

                client.on("ED718/downRaw/ty02A1/+/data", new MqttHandler() {
                    @Override
                    public void onMessage(String topic, ByteBuf payload) {
                        byte[] data = new byte[payload.readableBytes()];
                        payload.readBytes(data);
                        try {
                            String dd = new
                                    String(config.getHandler().decrypt(data));
                            System.out.println("==============================="+dd);
                        }catch (Exception e){
                            e.printStackTrace();
                        }


                    }
                });
                config.setClient(client);
                i++;
            }catch (Exception e){
                System.out.println("=============================="+i);
                e.printStackTrace();
            }
        }
        send(configs);


//        System.out.println(JSONObject.toJSONString(configs));
    }

    public static void send(List<ClientConfig> configs) throws InterruptedException {

        ExecutorService service = Executors.newFixedThreadPool(10);

        byte [] data = "{\"msgType\":\"deviceHeart\",\"deviceId\":\"08d1f9c95c6a\",\"cmd\":102,\"sn\":3,\"devType\":60006,\"ts\":1696904415,\"imei\":867793064104617,\"mac\":\"08d1f9c95c6a\",\"data\":\"ruvjG4ntTAyp0iS4O6VvfsJFyA1DO6DBk2N6H6MGB8uhhaaer+Rmpbsl0yHPs1ljP3Wx739zHfiNTt7M3SkaOs60MExugjxYwbgak6OZUx2JGcAmJXolMK3Ozz8RvnCxyHnKVldD+WUJJ7ptE6KHgjuvK/y7We1sqYCRzChVETI\\u003d\",\"rssi\":-63,\"iccid\":\"89860623550049327077\"}".getBytes(StandardCharsets.UTF_8);

        byte [] updata = "{\"cmd\":101,\"sn\":78,\"rssi\":-67,\"battery\":100,\"devType\":69001,\"imei\":\"869334050147312\",\"mac\":\"e89f6ddfcc6c\",\"ts\":1546270000,\"data\":\"0YQjLJqdwSjVRDWXFO7eAg==\",\"iccid\":\"8986062022007621019917\"}".getBytes(StandardCharsets.UTF_8);

        while (true){
            for (ClientConfig config :configs) {
                if(!config.getHandler().getReInit()){
                    continue;
                }
                service.submit(new Runnable() {
                    @Override
                    public void run() {
                        ByteBuf byteBuf = ALLOCATOR.buffer();
                        ByteBuf upbyteBuf = ALLOCATOR.buffer();
                        byte [] endata ;
                        byte [] upendata ;
                        try {
                            endata = config.getHandler().encrypt(data);
                            upendata = config.getHandler().encrypt(updata);
                        } catch (Exception e) {
                            throw new RuntimeException(e);
                        }
                        byteBuf.writeBytes(endata);
                        upbyteBuf.writeBytes(upendata);
                        config.getClient().publish("ED718/upRaw/08d1f9c95c6a/ts/data",byteBuf,MqttQoS.AT_MOST_ONCE);
                        config.getClient().publish("ED718/upRaw/08d1f9c95c6a/ts/data",upbyteBuf,MqttQoS.AT_MOST_ONCE);
                    }
                });
            }
            Thread.sleep(1000);
        }
    }

    public static List<ClientConfig> getDeviceList() {
        List<ClientConfig> clientConfigs = new ArrayList<>();
        String JDBC_URL = "jdbc:mysql://172.21.32.21:3306/tyiotcloud";
        String USER = "root";
        String PASSWORD = "!QAZ2wsx";
        String sql = "select credentials_value ,device_id FROM device LEFT JOIN device_credentials ON device.id = device_id WHERE type = '精华隆-ED718人体存在' limit 20 ";
        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
            return clientConfigs;
        }
        try (Connection connection = DriverManager.getConnection(JDBC_URL, USER, PASSWORD)) {
            System.out.println("Connected to the database");

            // 执行查询
            try (PreparedStatement preparedStatement = connection.prepareStatement(sql)) {

                // 执行查询
                try (ResultSet resultSet = preparedStatement.executeQuery()) {
                    System.out.println("Query result:");

                    while (resultSet.next()) {
                        String value = resultSet.getString("credentials_value");
                        String device_id = resultSet.getString("device_id");
                        ClientConfig config = JSONObject.parseObject(value, ClientConfig.class);
                        config.setDeviceId(device_id);
                        clientConfigs.add(config);
                    }
                }
            }
            connection.close();

        } catch (SQLException e) {
            e.printStackTrace();
        }

        return clientConfigs;
    }
}